/*
 * Replica server - two-phase commit protocol implementation
 * A thread named "replica" implements the two-phase commit protocol for any message that is to be stored
 * in the bulletin board (master) or for any message received from others servers in the peer list (slave).
 * If the master operation is successful, then the message is guaranteed to have been stored in all servers.
 * If the slave operation is successful, the bulletin board of the server running this thread is updated with
 * the new message. If either operation fails, all bulletin boards from all servers remain unchanged.
 */

#include "constants.h"
#include "config.h"
#include "utils.h"
#include "tcp-utils.h"
#include "bulletinboard.h"
#include "writebuffer.h"

#include <sstream>
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <ctime>
#include <unistd.h>
#include <syslog.h>

enum finite_machine  {IDLE, PRECOMMIT, COMMITED}; //States of the finite state machine
enum fsm_mode {SLAVE, MASTER, NEITHER}; //Replica thread can operate as slave, master or neither.

//Bulletin board file
extern BulletinBoard bb;
//Variables used for restarting or shutting server down
extern int shutdownStage;
extern pthread_mutex_t shutdownStageMutex;

// Updates own bulletin board
int update_board(const char *msg)
{
    log(LOG_DEBUG, "Update board with: %s\n", msg);
    int result = bb.writeMessage(msg);
    if ( result < 0 )
        log(LOG_DEBUG, "Update board failed: %d\n", result);
    return result;
}

// Undoes last operation
int undo()
{
    return bb.undoLastWrite();
}

// Broadcasts a message to all servers in the distributed network.
void broadcast(const char * msg, const vector<int> & sockets)
{
    log(LOG_DEBUG, "Broadcasting: %s", msg);
    for ( int i = 0; i < (int)sockets.size(); ++i ) //Sends message to all peers
    {
        int err = send( sockets[i], msg, strlen(msg), 0 );
        log(LOG_DEBUG, "Broadcasting to socket %d\n", sockets[i]);

        if ( err == -1 )
            logError(LOG_DEBUG, "Broadcast failed");
    }
}

// Answers one particular server.
void answer(const char *msg, int seq, int socket)
{
    char answer_msg[maxMessageLength+15];
    sprintf(answer_msg, "%d/%s\n", seq, msg);
    int err = send( socket, answer_msg, strlen(answer_msg), 0 );

    log(LOG_DEBUG, "Answer: %s\n", answer_msg);
    if ( err == -1 )
        logError(LOG_DEBUG, "Answer failed");
}

// Broadcasts a failure message and resets the master finite state machine  to the initial state (IDLE).
int fail(const char *msg, int seq, const vector<int> & list, finite_machine *state)
{
    char broadcast_msg[maxMessageLength+15];
    sprintf(broadcast_msg, "%d/%s\n", seq, msg);
    broadcast(broadcast_msg, list);
    *state = IDLE;
    return -1;
}

// If difference is greater than timeout, return true
int timeout (time_t a, time_t b, int timeout)
{
    return b-a > timeout;
}

// Converts IP integer to IP string (IPv4)
void print_ip(int ip, char *ip_str)
{
    unsigned char bytes[4];
    bytes[0] = ip & 0xFF;
    bytes[1] = (ip >> 8) & 0xFF;
    bytes[2] = (ip >> 16) & 0xFF;
    bytes[3] = (ip >> 24) & 0xFF;
    sprintf(ip_str, "%d.%d.%d.%d", bytes[3], bytes[2], bytes[1], bytes[0]);
}


// Master finite state machine - uses the three-way handshake algorithm to synchronize all servers in the distributed network with some data.
int master(int seq, const char *message, const vector<int> & socket_list)
{
    static finite_machine state = IDLE; //Initial state is idle
    static int internal_seq = rand() % replicaMaxSequence; //Internal sequence number used to synchronize messages
    static char buffer_msg[maxMessageLength], broadcast_msg[maxMessageLength]; //Buffer is the raw data to be sent - broadcast is the actual message that will be sent
    static int available_servers = 0, successful_servers = 0;
    static time_t time_start;

    switch(state)
    {
    case IDLE: //Initial state
        available_servers = successful_servers = 0; //Resets counters
        strncpy(buffer_msg, message, maxMessageLength); //Stores data from the buffer
        buffer_msg[maxMessageLength] = '\0';

        ++internal_seq; //Increase sequence number
        if ( internal_seq >= replicaMaxSequence )
            internal_seq = 0;

        sprintf(broadcast_msg, "%d/PRECOMMIT\n",internal_seq); //Broadcasts precommit message to all servers
        broadcast(broadcast_msg, socket_list);
        state = PRECOMMIT; //Goes to next state
        time_start = time(NULL); //Start counting time
        break;
    case PRECOMMIT: //Precommit state
        if  (timeout(time_start, time(NULL), replicaTimeout))  //Servers took too long to answer - operation must end
        {
            fail("ABORT", internal_seq, socket_list, &state);
            return -6;
        }

        if (seq == internal_seq)
        {
            if (strcmp(message, "AVAILABLE") == 0) //One server acknowledges us
                available_servers++;
            else if (strcmp(message, "UNAVAILABLE") == 0)   //One server is unavailable - operation must end
            {
                fail("ABORT", internal_seq, socket_list, &state);
                return -6;
            }
            if (available_servers == (int)socket_list.size())   //All servers are available - broadcast (commit) data
            {
                state = COMMITED;
                sprintf(broadcast_msg,"%d/COMMIT/%s\n", internal_seq, buffer_msg);
                broadcast(broadcast_msg, socket_list);
                time_start = time(NULL);
                break;
            }
        }

        break;
    case COMMITED: //Commit state
        if  (timeout(time_start, time(NULL), replicaTimeout))  //Servers took too long to answer - operation must end
        {
            fail("UNSUCCESSFUL", internal_seq, socket_list, &state);
            return -6;
        }
        if (seq == internal_seq)
        {
            if (strcmp(message, "SUCCESS") == 0) //One server received and committed the data.
                successful_servers++;
            else if (strcmp(message, "UNSUCCESS") == 0)  //One server couldn't commit the data - all servers must undo their actions
            {
                fail("UNSUCCESSFUL", internal_seq, socket_list, &state);
                return -6;
            }

            if (successful_servers == (int)socket_list.size())
            {
                int result = update_board(buffer_msg); //Updates own bulletin board status
                if( result >= 0 ) //Bulletin board updated successfully - information broadcasted to all other servers.
                    fail("SUCCESSFUL", internal_seq, socket_list, &state); //NOT a fail actually, but a successful broadcast that returns state to IDLE.
                else   //Bulletin board did not update successfully - all servers must undo their actions.
                    fail("UNSUCCESSFUL", internal_seq, socket_list, &state);
                return result;
            }
        }

        break;
    }
    return 0;
}

// Slave finite state machine - uses the three-way handshake algorithm to synchronize all servers in the distributed network with some data.
int slave(int seq, char *message, int socket)
{
    static finite_machine state = IDLE; //Initial state is idle
    static int internal_seq; //Internal sequence number used to synchronize messages
    char * buffer_msg;
    static time_t time_start;
    int error;

    switch(state)
    {
    case IDLE: //Initial state
        internal_seq = seq;
        answer("AVAILABLE", seq, socket);
        state = PRECOMMIT; //Goes to next state
        time_start = time(NULL);
        break;
    case PRECOMMIT: //Precommit state
        if  (timeout(time_start, time(NULL), replicaTimeout))  //Servers took too long to answer - operation must end
        {
            state = IDLE;
            return -1;
        }

        if (internal_seq == seq)
        {
            buffer_msg = strtok(message, "/");
            if (strcmp(buffer_msg, "COMMIT") == 0) //Receives the commit information
            {
                buffer_msg = strtok(NULL, "\n"); //Parses message
                buffer_msg[strlen(buffer_msg)] = '\0';

                if ((error = update_board(buffer_msg)) > 0) //Tries to update own board with the new message
                {
                    answer("SUCCESS", seq, socket); //If it succeeded, answer peer positively
                    state = COMMITED;
                    time_start = time(NULL);
                    break;
                }
                else //If it could not update board, answer negatively
                {
                    answer("UNSUCCESS", seq, socket);
                    state = IDLE;
                    return -1;
                }
            }
            else if (strcmp(buffer_msg, "ABORT") == 0)  //If it receives abort message, abort commit operation
            {
                state = IDLE;
                return -1;
            }
        }
        break;
    case COMMITED: //Commit state
        if  (timeout(time_start, time(NULL), replicaTimeout))  //Servers took too long to answer - operation must end
        {
            undo();
            state = IDLE;
            return -1;
        }
        if (seq == internal_seq) //Receives answer from the master regarding current commit
        {
            if (strcmp(message, "SUCCESSFUL") == 0) //Commit is done successfully
            {
                state = IDLE;
                return 1;
            }
            else if (strcmp(message, "UNSUCCESSFUL") == 0) //Commit was not successful so its message is removed from the board
            {
                undo();
                state = IDLE;
                return -1;
            }
        }
        break;
    }
    return 0;
}

// Thread that handles the input and activity of the master and the slave finite state machines.
void * replica_thread(void *)
{
    fsm_mode mode = NEITHER; //Mode of the finite state machine
    char msg[maxMessageLength], ip[16]; //msg can be either a message read from buffer or from one peer - ip is a IP string
    char garbage[]= "GARBAGE/GARBAGE"; //garbage message used to wake up master or slave so they can check for time out
    char *tok; //Used for string tokenization
    int seq, status = 0, newMessage = 0;
    vector<int> peerSockets, mapSocketsIPs; //Vector of the sockets of connected peers, vector of their indices in the IP list
    struct pollfd * fds = NULL;
    int last_read_socket = -1;

    //Creates and binds a master socket the it can connect to or receive the peers.
    int msock = master_socket(Config::self()->replicaPort());
    if (msock == -1)
    {
        logError(LOG_CRIT, "Could not create the master socket for the replica thread");
        exit(1);
    }

    vector<string> peer_list = Config::self()->peers();

    //Tries to connect to all peers on the peer list.
    for (int i = 0; i < (int) peer_list.size(); i++)
    {
        int sock = connectbyportint(peer_list[i].c_str(), Config::self()->replicaPort());
        if ( sock > 0) //Succeeded connecting to peer
        {
            peerSockets.push_back( sock );
            mapSocketsIPs.push_back( i );
            log(LOG_DEBUG, "Connected to peer %s. \n", peer_list[i].c_str());
        }
        else //Failed connecting to peer (this peer can still connect to us)
            log(LOG_WARNING, "Could not connect to peer %s. \n", peer_list[i].c_str());
    }

    while( true )
    {
        int numFds = peerSockets.size()+1; //Number of connected peers

        //Creates information for all peers and master socket so they can all be polled together
        if (fds == NULL)
        {
            fds = new struct pollfd [ numFds ];

            for (int i = 0; i < numFds; i++)
            {
                fds[i].revents = 0;
                fds[i].events = POLLIN; //Incoming information
                if (i == 0)
                    fds[i].fd = msock;
                else
                    fds[i].fd = peerSockets[i-1];
            }
        }
        //Poll for incoming connection for TIMEOUT ms.
        int poll_result = poll(fds, (nfds_t) numFds, pollTimeout);
        if (poll_result > 0)
        {
            for (int i = 0; i < numFds; i++)
            {
                if (fds[i].revents != 0) //If there is incoming information
                {
                    if (i == 0) //If it is in the master socket, creates new connection if valid peer
                    {
                        struct sockaddr_in client_addr; //The address of the client
                        socklen_t client_addr_len = sizeof(client_addr); //Client's length
                        int newSock = accept(msock, (struct sockaddr*)&client_addr, &client_addr_len); //If there is a pending connection, accepts it.
                        if ( newSock > 0 )
                        {
                            print_ip(htonl(client_addr.sin_addr.s_addr) , ip);
                            log(LOG_DEBUG, "Peer %s tried to connect to the server.\n", ip);

                            //Verifies peer validity
                            for (i = 0; i < (int)peer_list.size(); i++)
                            {
                                struct hostent * hostDataAux = gethostbyname(peer_list[i].c_str());
                                struct sockaddr_in aux;
                                memcpy(&aux.sin_addr, hostDataAux->h_addr, hostDataAux->h_length);
                                //If peer is valid, accept new connection and update socket list.
                                if (htonl(client_addr.sin_addr.s_addr) == htonl((aux.sin_addr.s_addr)))
                                {
                                    log(LOG_DEBUG, "Connection accepted.\n");
                                    peerSockets.push_back( newSock );
                                    mapSocketsIPs.push_back( i );
                                    delete [] fds; //Forces poll vector to be rebuilt
                                    fds = NULL;
                                    break;
                                }
                            }
                            if (i == (int)peer_list.size()) //Connection is not a valid peer
                            {
                                log(LOG_DEBUG, "Connection refused - could not find peer on peer list.\n");
                                shutdown(newSock, SHUT_RDWR);
                                close(newSock);
                            }
                        }
                    }
                    else //Incoming information is from a peer
                    {
                        int len = readline(peerSockets[i-1], msg, maxMessageLength);
                        if (len > 0) //Reads its message
                        {
                            msg[len] = '\n';
                            msg[len+1] = '\0';
                            log(LOG_DEBUG, "Message received: %s", msg);
                            newMessage = 1; //Message shall be acknowledged by the master protocol
                            last_read_socket = peerSockets[i-1]; //Saves peer socket so the slave can answer directly to it
                        }
                        if (len < 0) //If peer disconnected, updates socket list.
                        {
                            log(LOG_DEBUG, "Peer %s disconnected.\n", peer_list[mapSocketsIPs[i-1]].c_str());
                            shutdown(peerSockets[i-1], SHUT_RDWR);
                            close(peerSockets[i-1]);
                            for (int j = i - 1; j < (int)peerSockets.size() - 1; j++)
                            {
                                peerSockets[j] = peerSockets[j+1];
                                mapSocketsIPs[j] = mapSocketsIPs[j+1];
                            }
                            peerSockets.pop_back();
                            mapSocketsIPs.pop_back();
                            delete [] fds;
                            fds = NULL;
                        }
                    }
                    break;
                }
            }
        }
//If there is a message in the server buffer (i.e. a local client tried to write something)
        if ( WriteQueue::self()->hasMessageWaiting() )
        {
            if (newMessage == 0 && mode == NEITHER) //If there is no new message from the peers and master and slave are idle...
            {
                string aux = WriteQueue::self()->currentMessage();
                if (aux.substr(0, 3) == "-1/")   //Updates message with current message ID
                {
                    ostringstream oss;
                    oss << bb.lastMessageId() << aux.substr(2);
                    aux = oss.str();
                }
                if ( peerSockets.empty() ) //If no peer is connected, updates board right way without the two-phase commit protocol
                {
                    strncpy(msg, aux.c_str(), aux.size());
                    int result = update_board( msg );
                    WriteQueue::self()->doneWithMessage( result );
                }
                else //Message shall be acknowledged by the slave protocol
                {
                    //"1/" is just a garbage sequence number that will be tokenized out of the string later on.
                    aux = "1/" + aux;
                    strncpy(msg, aux.c_str(), aux.size());
                    msg[aux.size()] = '\n';
                    newMessage = 2;
                }
            }
        }

        if (newMessage > 0) //Pending message
        {
            //Tokenizes message to get its sequence number out
            tok = strtok(msg, "/");
            if (strlen(tok) > 0)
                seq = atoi(tok);
            else
            {
                newMessage = 0;
                continue;
            }
            tok = strtok(NULL, "\n");
            if (strcmp(tok, "PRECOMMIT") == 0) //Received a precommit information
                if (mode != NEITHER) //If the server is not idle, it must answer as unavailable to requesting peer.
                    answer("UNAVAILABLE", seq, last_read_socket);
                else //Acknowledges precommit and begins slave mode
                    mode = SLAVE;
            else if (mode == NEITHER && newMessage == 2) //Message pending from own buffer, server will try to commit it to its peers
                mode = MASTER;

            if (mode == SLAVE)
            {
                status = slave(seq, tok, last_read_socket);
            }
            else if (mode == MASTER)
            {
                status = master(seq, tok, peerSockets);
                if ( status != 0 ) //Commit protocol either succeeded or failed
                    //Commit status is returned to the buffer so it can delete it or try to send it again (depending on the answer from the commit procotol)
                    WriteQueue::self()->doneWithMessage( status );
            }

            if (status != 0) //Commit protocol from slave or master is over, thread is idle again
                mode = NEITHER;

            newMessage = 0; //Pending information has been dealt
        }
        else
        {
            //In case there is no pending information, master and slave are sent garbage information so they can check out for timeout.
            if (mode == MASTER)
            {
                status = master(-1, (const char *)garbage, peerSockets);
                if ( status != 0 ) //If commit timed out, the buffer must be informed of that
                    WriteQueue::self()->doneWithMessage( status );
            }
            else if (mode == SLAVE)
            {
                status = slave(-1, garbage, last_read_socket);
            }
            if (status != 0) //If it timed out, this thread is idle again
                mode = NEITHER;
        }
        //Checks for shutdown signal
        pthread_mutex_lock( &shutdownStageMutex );
        bool gottaGo = shutdownStage == 3;
        pthread_mutex_unlock( &shutdownStageMutex );
        if ( gottaGo )
            break;
    }
//If signal to shutdown has been sent, thread must be terminated
    shutdown( msock, SHUT_RDWR );
    close( msock );
    for ( int i = 0; i < (int)peerSockets.size(); ++i )
    {
        shutdown( peerSockets[i], SHUT_RDWR );
        close( peerSockets[i] );
    }
    delete [] fds;

    return NULL;
}

